/*
 * @Descripttion: mqtt测试文件，包含了连接订阅发布等函数的封装
 * @version: 
 * @@Company: 
 * @Author: Crystal
 * @Date: 2021-04-04 09:46:56
 * @LastEditors: Crystal
 * @LastEditTime: 2021-04-04 14:33:40
 */

#include <stdint.h>
#include <stdio.h>
#include <string.h>
#include <pthread.h>
#include <unistd.h>
#include <time.h>
#include <signal.h>

#include "MQTTPacket.h"
#include "MQTTConnect.h"
#include "transport.h"
#include "mqtt_test.h"

#define LOG_TAG "mqtt_test"
#include "elog.h"
#include "elog_start.h"

static pthread_t pth_recv;
static pthread_t pth_heartbeat;                   // 心跳
static uint8_t mqtt_alive_status = MQTT_NOT_CONN; // 定义连接状态
static timer_t mqtt_heartbeat_timer = NULL;       // 心跳定时器
static pthread_mutex_t mutex;                     // 锁

/*************************  连接相关  *************************/

/**
 * @description: tcp连接
 * @param   void
 * @return  成功返回0，失败返回-1
 * @author: Crystal
 */
int tcp_connect(void)
{
    int socket;

    socket = transport_open(SERVER_ADDR, SERVER_PORT); // 创建一个socket，和服务器连接
    if (socket < 0)
    {
        log_e("transport_open failed\n");
        return -1;
    }
    else
    {
        log_i("tcp connect success\n");
    }

    return socket;
}

/**
 * @description: 连接MQTT服务器
 * @param  void
 * @return 成功返回0，失败返回-1
 * @author: Crystal
 */
int mqtt_conn(void)
{
    MQTTPacket_connectData options = MQTTPacket_connectData_initializer; // 连接
    uint8_t send_buf[256] = {0};
    int send_len;
    uint8_t recv_buf[8] = {0};
    int recv_cnt;
    int ret = 0, i;

    options.keepAliveInterval = HEARTBEAT_SEC; // 心跳包时间
    options.cleansession = 1;                  // 客户端和服务端 必须丢弃之前的任何会话并开始一个新的会话
    options.willFlag = 0;                      // 遗嘱标志

    options.clientID.cstring = CLIENT_ID; // 腾讯云为: 产品 ID + 设备名
    options.username.cstring = USER_NAME;
    options.password.cstring = USER_PASSWORD;

    send_len = MQTTSerialize_connect(send_buf, sizeof(send_buf), &options); // 组织连接报文
    if (send_len < 0)
    {
        log_e("MQTTSerialize_connect error, send_len = %d\n", send_len);
        return -1;
    }

    // 打印发送的内容
    log_v("send_len:%d\n", send_len);
    for (i = 0; i < send_len; ++i)
        log_v("0x%02x ", send_buf[i]);
    log_v("\n");

    ret = transport_sendPacketBuffer(send_buf, send_len); // 向MQTT服务器发送连接请求
    if (ret < 0)
    {
        log_e("send failed, ret = %d\n", ret);
        return -1;
    }

    // 不断重连
    while (true)
    {
        if (MQTTPacket_read(recv_buf, sizeof(recv_buf), transport_getdata) == CONNACK) //CONNACK – 确认连接请求
        {
            uint8_t sessionPresent, connack_rc;

            // 反序列化收到的内容, sessionPresent为连接确认标志, connack_rc为连接返回码
            if (MQTTDeserialize_connack(&sessionPresent, &connack_rc, recv_buf, sizeof(recv_buf)) != 1 || connack_rc != MQTT_CONNECTION_ACCEPTED)
            {
                mqtt_alive_status = MQTT_CONN_FAILED; // 将连接状态更改为连接失败
                log_e("Unable to connect, return code %d\n", connack_rc);
            }
            else
            {
                mqtt_alive_status = MQTT_CONN_SUCC; // 将连接状态更改为连接成功
                log_i("mqtt connect server success\n");
                return 0;
            }
        }

        sleep(2);                             // 连接失败
        mqtt_alive_status = MQTT_CONN_FAILED; // 将连接状态更改为连接失败
        ret = transport_sendPacketBuffer(send_buf, send_len);
        log_i("retry conn mqtt server\n");
        if (ret < 0)
        {
            log_e("send failed, ret = %d\n", ret);
            return -1;
        }
    }

    return 0;
}

/*************************  心跳包相关  *************************/

/**
 * @description: mqtt发送心跳包 
 * @param   void
 * @return  成功返回0，失败返回-1
 * @author: Crystal
 */
int mqtt_send_heartbeat(void)
{
    uint8_t send_buf[8] = {0};
    uint8_t send_len, i, ret;
    uint8_t recv_buf[8];

    send_len = MQTTSerialize_pingreq(send_buf, sizeof(send_buf) / sizeof(send_buf[0])); // 组织心跳包报文
    if (send_len <= 0)
    {
        log_e("MQTTSerialize_pingreq error, err:%d\n", send_len);
        return -1;
    }

    // 打印心跳包的报文内容
    log_v("\nheartbeat packet: ");
    for (i = 0; i < send_len; ++i)
    {
        log_v("0x%02x ", send_buf[i]);
    }
    log_v("\n");

    pthread_mutex_lock(&mutex); // 上锁

    ret = transport_sendPacketBuffer(send_buf, send_len);
    if (ret <= 0)
    {
        log_e("send heart beat failed\n");
        pthread_mutex_unlock(&mutex); // 解锁
        return -1;
    }

    if (MQTTPacket_read(recv_buf, sizeof(recv_buf) / sizeof(recv_buf[0]), transport_getdata) == PINGRESP) // 心跳包响应报文
    {
        log_d("send heartbeat success\n");
    }

    pthread_mutex_unlock(&mutex); // 解锁

    return 0;
}

/**
 * @description: mqtt心跳线程处理函数
 * @param {void} *p
 * @return {*}
 * @author: Crystal
 */
void mqtt_heartbeat_timer_thread_handle(union sigval v)
{
    int i, j;
    uint8_t recv_buf[8] = {0};

    mqtt_heartbeat_timer_stop();

    if (mqtt_alive_status == MQTT_CONN_SUCC) // 如果现在的状态是连接，则直接发送心跳包
    {
        for (i = 0; i < MQTT_SEND_CNT; ++i)
        {
            if (mqtt_send_heartbeat() == 0) // 发送心跳包成功
            {
                mqtt_heartbeat_timer_start(HEARTBEAT_SEC / 2); // 重新启动定时器
                return;
            }
        }

        if (i == MQTT_SEND_CNT) // 发送了MQTT_SEND_CNT 仍然未成功，说明连接有问题
        {
            mqtt_alive_status == MQTT_CONN_FAILED;         // 讲状态修改为连接失败
            mqtt_heartbeat_timer_start(HEARTBEAT_SEC / 3); // 重新启动定时器
        }
    }
    else // 连接失败或者未连接需要重新建立连接
    {
        if (mqtt_conn() == 0) // 连接失败重新连接，重新连接如果成功
        {
            mqtt_alive_status == MQTT_CONN_SUCC;           // 讲状态修改为连接失败
            mqtt_heartbeat_timer_start(HEARTBEAT_SEC / 2); // 重新启动定时器
        }
        else
        {
            mqtt_heartbeat_timer_start(HEARTBEAT_SEC / 3); // 重新启动定时器
        }
    }
}

/**
 * @description: 启动心跳线程
 * @param  void
 * @return void
 * @author: Crystal
 */
void start_mqtt_heartbeat_thread(void)
{
    if (mqtt_heartbeat_timer_init() != 0) // 初始化一个定时器
        return;

    mqtt_heartbeat_timer_start(HEARTBEAT_SEC / 2); // 启动定时器
}

/**
 * @description: 初始化心跳包定时器 
 * @param   void
 * @return  成功返回0，失败返回-1
 * @author: Crystal
 */
int mqtt_heartbeat_timer_init(void)
{
    int ret = -1;

    struct sigevent evp;                      // 参数evp指定了定时器到期要产生的异步通知
    memset(&evp, 0, sizeof(struct sigevent)); // 清零初始化

    evp.sigev_value.sival_int = 111;                                //也是标识定时器的，这和timerid有什么区别？回调函数可以获得
    evp.sigev_notify = SIGEV_THREAD;                                // 当定时器到期，派驻新线程
    evp.sigev_notify_function = mqtt_heartbeat_timer_thread_handle; //线程函数地址

    ret = timer_create(CLOCK_REALTIME, &evp, &mqtt_heartbeat_timer);
    if (ret != 0)
        log_e("mqtt heartbeat timer create failed\n");

    return ret;
}

/**
 * @description: 启动心跳包定时器 
 * @param   void
 * @return  void
 * @author: Crystal
 */
void mqtt_heartbeat_timer_start(int heartbeat_sec)
{
    int ret;
    struct itimerspec ts;

    ts.it_interval.tv_sec = 0; // 当定时器到期，it_value的值会被更新成it_interval 的值
    ts.it_interval.tv_nsec = 0;
    ts.it_value.tv_sec = heartbeat_sec; // 用于指定当前的定时器到期时间
    ts.it_value.tv_nsec = 0;

    ret = timer_settime(mqtt_heartbeat_timer, 0, &ts, NULL);
    if (ret != 0)
        log_e("timer settime failed\n");
}

/**
 * @description: 停止心跳包定时器 
 * @param   void
 * @return  void
 * @author: Crystal
 */
void mqtt_heartbeat_timer_stop(void)
{
    int ret;
    struct itimerspec ts;

    ts.it_interval.tv_sec = 0; // 当定时器到期，it_value的值会被更新成it_interval 的值
    ts.it_interval.tv_nsec = 0;
    ts.it_value.tv_sec = 0; // 用于指定当前的定时器到期时间
    ts.it_value.tv_nsec = 0;

    ret = timer_settime(mqtt_heartbeat_timer, 0, &ts, NULL);
    if (ret != 0)
        log_e("timer settime failed\n");
}

/**
 * @description: 删除mqtt心跳包定时器
 * @param   void
 * @return  void
 * @author: Crystal
 */
void mqtt_heartbeat_timer_delete(void)
{
    if (mqtt_heartbeat_timer != NULL)
    {
        timer_delete(mqtt_heartbeat_timer);
        mqtt_heartbeat_timer = NULL;
        log_i("delete mqtt heartbeat timer\n");
    }
}

/*************************  发布相关  *************************/

/**
 * @description: mqtt发布消息至服务器
 * @param   uint8_t *payload 要发布的消息的内容
 * @param   int payloadlen   要发布的消息的长度
 * @return  成功返回0，失败返回-1
 * @author: Crystal
 */
int mqtt_publish_msg(uint8_t *payload, int payloadlen)
{
    MQTTString topicString = MQTTString_initializer;
    uint8_t send_buf[1024] = {0};
    int send_len;
    int i, ret;
    int qos = QOS_GRADE;
    uint8_t dup = 0;
    uint8_t recv_buf[8] = {0};
    int recv_len;

    topicString.cstring = PUBLISH_TOPIC_NAME; // 发布的主题名字

    // 组消息帧
    send_len = MQTTSerialize_publish(send_buf, sizeof(send_buf), dup, qos, RETAIN_FALG, PACKET_IDENTIFIER,
                                     topicString, payload, payloadlen);

    // 打印发布的消息的数据帧内容
    log_v("\npublish packet: ");
    for (i = 0; i < send_len; ++i)
        log_v("0x%02x ", send_buf[i]);
    log_v("\n");

    if (qos == 0) //qos 响应报文，当qos>0时才有响应
        return 0;

    pthread_mutex_lock(&mutex); // 上锁

    ret = transport_sendPacketBuffer(send_buf, send_len); // 发送消息
    if (ret <= 0)
    {
        log_e("publish msg failed\n");
        goto err0;
    }

    if (MQTTPacket_read(recv_buf, sizeof(recv_buf), transport_getdata) == PUBACK) //qos 响应报文，当qos>0时才有响应
    {
        // 打印收到的发布消息的回应数据包
        log_v("publish recv packet: ");
        for (i = 0; i < sizeof(recv_buf); ++i)
            log_v("0x%02x ", recv_buf[i]);
        log_v("\n");

        if (recv_buf[1] != 0x02 || ((recv_buf[2] << 8) | recv_buf[3] != PACKET_IDENTIFIER))
        {
            log_e("mqtt recv publish ack failed\n");
            goto err0;
        }
        else
        {
            log_i("mqtt publish success. payload is %s\n", payload);
            ret = 0;
        }
    }
    else
    {
        // 打印收到的发布消息的回应数据包
        log_v("publish recv packet: ");
        for (i = 0; i < sizeof(recv_buf); ++i)
            log_v("0x%02x ", recv_buf[i]);
        log_v("\n");
        log_e("mqtt recv publish ack failed\n");
        goto err0;
    }

    pthread_mutex_unlock(&mutex); // 解锁

    return ret;

err0:
    pthread_mutex_unlock(&mutex); // 解锁
    return -1;
}

/*************************  订阅相关  *************************/

/**
 * @description: MQTT订阅主题
 * @param   void
 * @return  成功返回0，失败返回-1
 * @author: Crystal
 * @note    订阅腾讯云订阅两个主题时只能订阅成功一个，还不知道原因
 */
int mqtt_subscribe_msg(void)
{
    MQTTString topicStrings[SUBCRIBE_TOPIC_CNT] = {MQTTString_initializer, MQTTString_initializer}; // 可以同时订阅多个主题
    uint8_t send_buf[256];
    int req_qoss[SUBCRIBE_TOPIC_CNT] = {QOS_GRADE, QOS_GRADE};
    int send_len, ret, i, dup = 0;
    uint8_t recv_buf[8];
    int recv_len;

    topicStrings[0].cstring = SUBCRIBE_TOPIC_1_NAME;
    topicStrings[1].cstring = SUBCRIBE_TOPIC_2_NAME;

    send_len = MQTTSerialize_subscribe(send_buf, sizeof(send_buf) / sizeof(send_buf[0]), dup,
                                       SUBCRIBE_TOPIC_IDENTIFIER, SUBCRIBE_TOPIC_CNT, topicStrings, req_qoss);

    // 打印订阅的消息的数据帧内容
    log_v("\nsubscribe send packet: ");
    for (i = 0; i < send_len; ++i)
        log_v("0x%02x ", send_buf[i]);
    log_v("\n");

    ret = transport_sendPacketBuffer(send_buf, send_len); // 发送消息
    if (ret <= 0)
    {
        log_e("subscribe topic failed\n");
        goto err0;
    }

    if (MQTTPacket_read(recv_buf, sizeof(recv_buf), transport_getdata) == SUBACK)
    {
        uint16_t submsgid;
        int subcount, granted_qos[SUBCRIBE_TOPIC_CNT];
        ret = MQTTDeserialize_suback(&submsgid, SUBCRIBE_TOPIC_CNT, &subcount, granted_qos, recv_buf, sizeof(recv_buf));
        if (ret != 1) // 成功返回1
        {
            log_e("MQTTDeserialize_suback failed\n");
            goto err0;
        }

        // 打印收到的订阅消息的回应数据包
        log_v("subscribe recv packet: ");
        for (i = 0; i < sizeof(recv_buf); ++i)
            log_v("0x%02x ", recv_buf[i]);
        log_v("\n");
        log_i("mqtt subscribe success\n");
        return 0;
    }
    else
    {
        // 打印收到的订阅消息的回应数据包
        log_v("subscribe recv packet: ");
        for (i = 0; i < sizeof(recv_buf); ++i)
            log_v("0x%02x ", recv_buf[i]);
        log_v("\n");
        log_e("recv subscribe ack failed\n");
        goto err0;
    }

err0:
    return -1;
}

/*************************  接收相关  *************************/

/**
 * @brief  解析收到的数据
 * @note   
 * @param  *recv_buf: 
 * @param  recv_buflen: 
 * @retval 成功返回0，失败返回-1
 */
int parse_recv_data(uint8_t *recv_buf, int recv_buflen)
{
    uint8_t dup, retained;
    uint16_t packetid;
    int qos;
    MQTTString topicName = MQTTString_initializer;
    uint8_t *payload = NULL;
    int payloadlen = 0;
    int ret;
    char name[128] = {0};

    // 解析收到的数据
    ret = MQTTDeserialize_publish(&dup, &qos, &retained, &packetid, &topicName, &payload, &payloadlen, recv_buf, recv_buflen);
    if (ret != 1) // 成功返回1
    {
        log_e("MQTTDeserialize_publish failed, errno:%d\n", ret);
        return -1;
    }

    strncpy(name, topicName.lenstring.data, topicName.lenstring.len);
    log_d("topicname: %s\n", name);
    log_d("payload:   %s\n", payload);
    log_d("payloadlen:%d\n", payloadlen);

    return 0;
}

/**
 * @description: MQTT接收处理线程
 * @param   void *p
 * @return  void *
 * @author: Crystal
 */
void *mqtt_recv_handle(void *p)
{
    int recv_num = -1;
    uint8_t recv_buf[256] = {0};
    int count = sizeof(recv_buf), i;

    while (1)
    {
        pthread_mutex_lock(&mutex); // 上锁
        recv_num = transport_getdata(recv_buf, count);
        pthread_mutex_unlock(&mutex); // 解锁

        if (recv_num <= 0)
        {

            sleep(1);
        }
        else
        {
            log_v("\nrecv thread buf:");
            for (i = 0; i < recv_num; ++i)
                log_v("0x%02x ", recv_buf[i]);
            log_v("\n");

            parse_recv_data(recv_buf, sizeof(recv_buf));
        }
    }
}

/**
 * @description: 启动MQTT接收线程
 * @param   void
 * @return  void
 * @author: Crystal
 */
void start_mqtt_recv_thread(void)
{
    pthread_mutex_init(&mutex, NULL); // 初始化一个互斥锁

    pthread_create(&pth_recv, NULL, mqtt_recv_handle, NULL);
    pthread_detach(pth_recv);
}

/*************************  取消订阅相关  *************************/

/**
 * @description: MQTT取消订阅
 * @param   void
 * @return  成功返回0，失败返回-1
 * @author: Crystal
 */
int mqtt_unsubscribe(void)
{
    uint8_t send_buf[256] = {0};
    int send_len, ret, i;
    MQTTString topicString[UNSUBCRIBE_TOPIC_CNT] = {UNSUBCRIBE_TOPIC_1_NAME};
    uint8_t dup = 0;

    uint8_t recv_buf[8] = {0};

    send_len = MQTTSerialize_unsubscribe(send_buf, sizeof(send_buf), dup, UNSUBCRIBE_TOPIC_IDENTIFIER,
                                         UNSUBCRIBE_TOPIC_CNT, topicString);
    if (send_len <= 0)
    {
        log_e("MQTTSerialize_unsubscribe error, errno:%d\n", send_len);
        goto err0;
    }

    // 打印取消订阅的消息的数据帧内容
    log_v("\nunsubscribe send packet: ");
    for (i = 0; i < send_len; ++i)
        log_v("0x%02x ", send_buf[i]);
    log_v("\n");

    pthread_mutex_lock(&mutex);

    ret = transport_sendPacketBuffer(send_buf, send_len); // 发送消息
    if (ret <= 0)
    {
        log_e("unsubscribe topic failed\n");
        goto err0;
    }

    if (MQTTPacket_read(recv_buf, sizeof(recv_buf), transport_getdata) == UNSUBACK)
    {
        uint16_t de_packetid;
        ret = MQTTDeserialize_unsuback(&de_packetid, recv_buf, sizeof(recv_buf));
        if (ret != 1) // 成功返回1
        {
            log_e("MQTTDeserialize_unsuback failed. errno:%d\n", ret);
            goto err0;
        }
        else
        {
            log_i("unsubscribe topic success. packetid:%d\n", de_packetid);
            ret = 0;
        }
    }
    else
    {
        // 打印收到的取消订阅的消息帧的内容
        log_v("recv packet: ");
        for (i = 0; i < sizeof(recv_buf); ++i)
            log_v("0x%02x ", recv_buf[i]);
        log_v("\n");
        log_e("recv unsubscribe ack failed\n");
        goto err0;
    }

    pthread_mutex_unlock(&mutex);
    return 0;

err0:
    pthread_mutex_unlock(&mutex);
    return -1;
}

/*************************  断开连接相关  *************************/

/**
 * @description: 断开MQTT的连接
 * @param   void
 * @return  成功返回0,失败返回-1
 * @author: Crystal
 */
int mqtt_disconnect(void)
{
    uint8_t send_buf[8];
    int send_len, ret;

    mqtt_heartbeat_timer_stop(); // 停止心跳定时器

    send_len = MQTTSerialize_disconnect(send_buf, sizeof(send_buf));
    if (send_len <= 0)
    {
        log_e("MQTTSerialize_disconnect failed. errno:%d\n", send_len);
        return -1;
    }

    pthread_mutex_lock(&mutex);
    ret = transport_sendPacketBuffer(send_buf, send_len); // 发送消息
    if (ret <= 0)
    {
        log_e("send disconnect packet failed\n");
        ret = -1;
    }
    else
    {
        log_i("send disconnect packet success\n");
        mqtt_alive_status = MQTT_NOT_CONN; // 重新定义连接状态
        ret = 0;
    }
    pthread_mutex_unlock(&mutex);

    return ret;

err0:
    return -1;
}

/**
 * @description: 回收资源防止内存泄露
 * @param   void
 * @return  void
 * @author: Crystal
 */
void recycle_resources(void)
{
    // 1. 回收定时器的资源
    mqtt_heartbeat_timer_delete();
    log_i("delete heartbeat timer\n");

    // 2. 停止接收线程的工作
    pthread_cancel(pth_recv);
    log_i("cancel recv pthread\n");

    // 3.回收互斥锁资源
    pthread_mutex_destroy(&mutex);
    log_i("destory mutex\n");
}

/**
 * @description: 断开TCP的连接
 * @param   void
 * @return  void
 * @author: Crystal
 */
void tcp_disconnect(void)
{
    recycle_resources(); // 回收资源
    transport_close();   // 关闭TCP连接
    log_i("close tcp connect\n");
}